package com.person.server;

import com.person.server.grpcjava.*;
import io.grpc.stub.StreamObserver;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class PersonServiceImpl extends  PersonServiceGrpc.PersonServiceImplBase {
    /**
     * 客户端非流式，服务端非流式
     * 服务器响应必须调用onCompleted，客户端才会接收到消息
     * @param request
     * @param responseObserver
     */
    @Override
    public void getRealNameByUsername(MyRequest request, StreamObserver<MyResponse> responseObserver) {
        System.out.println("getRealNameByUsername接收到的客户端非流式消息为:"+request.getUsername());
        //响应消息
        responseObserver.onNext(MyResponse.newBuilder().setRealname("公众号:霸道的程序猿1").build());
        //服务器响应必须调用onCompleted，客户端才会接收到消息
        responseObserver.onCompleted();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 客户端非流式，服务端流式实现
     * 服务器响应必须调用onCompleted，客户端才会接收到消息
     * 服务器可以发送多条哦
     * @param request
     * @param responseObserver
     */
    @Override
    public void getStudentsByAge(StudentRequest request, StreamObserver<StudentResponse> responseObserver) {
        System.out.println("接收到的客户端消息为:"+request.getAge());

        responseObserver.onNext(StudentResponse.newBuilder().setName("1公众号:霸道的程序猿")
                .setAge(30)
                .setCity("北京")
                .build());

        responseObserver.onNext(StudentResponse.newBuilder().setName("2公众号:霸道的程序猿")
                .setAge(40)
                .setCity("上海")
                .build());

        responseObserver.onNext(StudentResponse.newBuilder().setName("3公众号:霸道的程序猿")
                .setAge(50)
                .setCity("广州")
                .build());

        responseObserver.onCompleted();
    }


    /**
     * 客户端流式，服务端非流式
     * 客户端流式时，服务端实现StreamObserver接口，实现onNext方法用于获取请求消息内容
     * 客户端每发送一条，服务器就每收一条。
     * 客户端onCompleted后，服务端开始处理onCompleted方法
     */
    @Override
    public StreamObserver<StudentRequest> getStudentsWrapperByAges(final StreamObserver<StudentResponseList> responseObserver) {


        StreamObserver<StudentRequest> streamObserverImpl = new StreamObserver<StudentRequest>() {
            //客户端每调用onNext，服务端就接收一次
            @Override
            public void onNext(StudentRequest studentRequest) {
                System.out.println("onNext:"+studentRequest.getAge());
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }
            @Override
            public void onCompleted() {
                System.out.println("onCompleted");
                responseObserver.onCompleted();
            }
        };

        StudentResponse studentResponse = StudentResponse.newBuilder()
                .setName("公众号:霸道的程序猿")
                .setAge(20)
                .setCity("北京").build();
        StudentResponse studentResponse1 = StudentResponse.newBuilder()
                .setName("1公众号:霸道的程序猿")
                .setAge(30)
                .setCity("上海").build();

        StudentResponseList studentResponseList = StudentResponseList.newBuilder()
                .addStudentResponse(studentResponse).addStudentResponse(studentResponse1).build();
        responseObserver.onNext(studentResponseList);
        responseObserver.onCompleted();

        return streamObserverImpl;

//        return new StreamObserver<StudentRequest>() {
//            //客户端每调用onNext，服务端就接收一次
//            public void onNext(StudentRequest studentRequest) {
//                System.out.println("onNext:"+studentRequest.getAge());
//            }
//
//            public void onError(Throwable throwable) {
//                System.out.println(throwable.getMessage());
//            }
//            //客户端调用onCompleted后，服务端开始处理响应报文
//            public void onCompleted() {
//                StudentResponse studentResponse = StudentResponse.newBuilder()
//                        .setName("公众号:霸道的程序猿")
//                        .setAge(20)
//                        .setCity("北京").build();
//                StudentResponse studentResponse1 = StudentResponse.newBuilder()
//                        .setName("1公众号:霸道的程序猿")
//                        .setAge(30)
//                        .setCity("上海").build();
//
//                StudentResponseList studentResponseList = StudentResponseList.newBuilder()
//                        .addStudentResponse(studentResponse).addStudentResponse(studentResponse1).build();
//                responseObserver.onNext(studentResponseList);
//                responseObserver.onCompleted();
//
//            }
//        };
    }

    /**
     * 双向流式实现
     * 客户端流式，服务端流式
     * 客户端必须使用异步的stub
     * 客户端和服务端必须实现StreamObserver接口，用于获取响应消息内容
     * 客户端和服务端均无需onCompleted，即可发送消息
     */
    Map<String,StreamObserver<StreamResponse>> map =new  HashMap<String,StreamObserver<StreamResponse>>();

    @Override
    public StreamObserver<StreamRequest> biTalk(StreamObserver<StreamResponse> responseObserver) {

        StreamObserver<StreamRequest> streamObserverImpl = new StreamObserver<StreamRequest>() {
            //客户端每调用onNext，服务端就接收一次
            @Override
            public void onNext(StreamRequest streamRequest) {
                System.out.println("收到客户端消息"+streamRequest.getRequestInfo()+streamRequest.getClientId());
                //登录
                if (streamRequest.getRequestInfo().contains("login")) {
                    map.put(streamRequest.getClientId(),responseObserver);
                    responseObserver.onNext(StreamResponse.newBuilder().setResponseInfo(streamRequest.getRequestInfo()+"登录成功").build());
                    //心跳
                }else if(streamRequest.getRequestInfo().contains("heartCheck")){
                    responseObserver.onNext(StreamResponse.newBuilder().setResponseInfo(streamRequest.getClientId()+"服务端收到，心跳请求").build());
                    //客户端需要牌价
                }else if(streamRequest.getRequestInfo().contains("okPrice")){
                    sendMsgToClient("okPrice牌价推送",streamRequest.getClientId());
                    //客户端需要钱
                } else if(streamRequest.getRequestInfo().contains("okMoney")){
                    sendMsgToClient("okMoney发钱啦发钱啦",streamRequest.getClientId());
                }else{
                    responseObserver.onNext(StreamResponse.newBuilder().setResponseInfo(streamRequest.getRequestInfo()+"请求，服务端已收到").build());
                }
             }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }
            @Override
            public void onCompleted() {
                System.out.println("onCompleted");
                responseObserver.onCompleted();
            }
        };
        //异步发送消息

        return streamObserverImpl;

    }

    //双向流式，模拟服务端不断发送消息给客户端
    public void sendMsgToClient(String msg,String clientId){
        // 开启定时器，向客户端发送消息
        ScheduledExecutorService timer = Executors.newScheduledThreadPool(1);
        timer.scheduleWithFixedDelay(() -> {
            //lamba表达式写一个线程
            try {
                map.get(clientId).onNext(StreamResponse.newBuilder().setClientId(clientId).setResponseInfo(System.nanoTime() + msg).build());
                System.out.println("服务端给客户端"+clientId+"发送消息：");
            } catch (Exception e) {
                // 移除客户端
                timer.shutdownNow();
            }
        }, 0, 3, TimeUnit.SECONDS);
    }
}